package com.leilei.broadcaststate.common;

import com.rabbitmq.client.AMQP;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/24 21:43
 */
public class RabbitMqSource extends RMQSource {

    public String exchangeName;

    public RabbitMqSource(RMQConnectionConfig rmqConnectionConfig, String queueName, String exchangeName) {
        super(rmqConnectionConfig, queueName, new SimpleStringSchema(StandardCharsets.UTF_8));
        this.exchangeName = exchangeName;
    }

    @Override
    protected void setupQueue() throws IOException {
        AMQP.Queue.DeclareOk result = channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(result.getQueue(), exchangeName, "*");
    }
}